package org.example.sec07;

import com.seaway.models.sec07.FlowControlServiceGrpc;
import com.seaway.models.sec07.Output;
import com.seaway.models.sec07.RequestSize;
import io.grpc.stub.StreamObserver;
import org.example.sec06.TransferService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.stream.IntStream;
import java.util.stream.LongStream;

/**
 * @author Administrator
 */
public class FlowControlService extends FlowControlServiceGrpc.FlowControlServiceImplBase {
    private static final Logger LOG = LoggerFactory.getLogger(FlowControlService.class);

    @Override
    public StreamObserver<RequestSize> getMessage(StreamObserver<Output> responseObserver) {
        return new RequestHandler(responseObserver);
    }

    private static class RequestHandler implements StreamObserver<RequestSize> {
        private StreamObserver<Output> responseObserver;
        private Integer emitted;

        public RequestHandler(StreamObserver<Output> responseObserver) {
            this.responseObserver = responseObserver;
            this.emitted = 0;
        }

        @Override
        public void onNext(RequestSize requestSize) {
            IntStream.rangeClosed(emitted + 1, 100)
                    .limit(requestSize.getValue())
                    .forEach(i -> {
                        LOG.info("emitting {}", i);
                        responseObserver.onNext(Output.newBuilder().setValue(i).build());
                    });
            emitted = emitted + requestSize.getValue();
            if (emitted >= 100) {
                responseObserver.onCompleted();
            }
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onCompleted() {
            responseObserver.onCompleted();
        }
    }
}
